package com.zbmy.plugin.im.server.core.zookeeper.server;

import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.jboss.netty.util.internal.ConcurrentHashMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.druid.util.StringUtils;
import com.base.framework.utils.JsonUtils;
import com.zbmy.plugin.im.server.core.zookeeper.server.node.ServerDataCache;

public class ZKServiceStore {
	Logger logger = LoggerFactory.getLogger(ZKServiceStore.class);
	private Map<String, String> userIdToNodePathMap = new ConcurrentHashMap<String, String>();
	private Map<String, String> userIdToServerMap = new ConcurrentHashMap<String, String>();
	private final ReentrantLock lock = new ReentrantLock();
	private final Condition conditionNotConnect = lock.newCondition(); //
    private final Condition conditionConnected = lock.newCondition();
	private volatile boolean notConnected = true;
	private ZooKeeper zk;
    private static String registryAddress = "localhost:2181";
    public ZKServiceStore() {
        zk = connectServer();  // 连接上zk服务器
        if (zk != null) {
        	notConnected = false;
        	initCreateIndexNode(zk);  // 初始化主节点
            // 观察节点数据
            new Thread(new Runnable() {
				@Override
				public void run() {
					while(true){
						// 每隔5秒钟主动拉取zk
						watchNode();
						try {
							Thread.sleep(5*1000);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					}
				}
			}).start();
        }else{
        	notConnected = true;
        }
        new Thread(new Runnable() {
			@Override
			public void run() {
				while(true){
					logger.info("链接监听机制");
					if(zk == null || notConnected){
						// 自动重连
						zk = connectServer();  // 连接上zk服务器
						if(zk != null){
							notConnected = false;
						}
					}
					try {
						conditionNotConnect.await(1000, TimeUnit.MILLISECONDS);
						Thread.sleep(10*1000);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			}
		}).start();
    }
    
    /**
	 * 监听ZK节点的数据变化
	 * TODO(这里用一句话描述这个方法的作用)    
	 * @author 杨乐平    
	 * @Title: watchNode    
	 * @param zk 
	 * @Return: void 返回值
	 */
    private void watchNode() {
        try {
            List<String> nodeList = zk.getChildren(ServerDataCache.ZK_CHAT_USER_INDEX_NODE_PATH , new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    if (event.getType() == Event.EventType.NodeChildrenChanged) {
                        watchNode();
                    }else if (event.getState() != Event.KeeperState.SyncConnected) {
                    	notConnected = false;
                    	conditionConnected.signal();
                    }else {
                    	notConnected = true;
                    	conditionNotConnect.signal();
                    }
                    try {
                        if (event.getType()==EventType.NodeCreated){
                            System.out.println(event.getPath()+" created");
                        }
                        else if (event.getType()==EventType.NodeDataChanged){
                            System.out.println(event.getPath()+" updated");
                        }
                        else if (event.getType()==EventType.NodeDeleted){
                            System.out.println(event.getPath()+" deleted");
                        }
                        
                    } catch (Exception e) {
                    }
                }
            });
            System.err.println("监听，nodeList ： {}" + JsonUtils.writeValueAsString(nodeList));
            for (String node : nodeList) {
            	// 获取节点的数据变化
                byte[] bytes = zk.getData(ServerDataCache.ZK_CHAT_USER_INDEX_NODE_PATH + "/" + node, false, null);
                String jsonData = new String(bytes);
                System.out.println("跟踪:" + JsonUtils.writeValueAsString(jsonData));
                if(StringUtils.isEmpty(jsonData)){
                	continue;
                }
                UserServerInfo userServerInfo = JsonUtils.readValue(jsonData, UserServerInfo.class);
                if(userServerInfo == null){
                	continue;
                }
                userIdToNodePathMap.put(userServerInfo.getUserId(), "/zk-node/chatUser/" + node);
                userIdToServerMap.put(userServerInfo.getUserId(), userServerInfo.getServerId());
            }
        } catch (Exception e) {
        	e.printStackTrace();
        }
    }

    /**
     * 
     * 连接zookeeper服务器   
     * @author 杨乐平    
     * @Title: connectServer    
     * @return 
     * @Return: ZooKeeper 返回值
     */
    private ZooKeeper connectServer() {
    	CountDownLatch latch = new CountDownLatch(1);
        ZooKeeper zk = null;
        try {
           zk = new ZooKeeper(registryAddress, 20000, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    if (event.getState() == Event.KeeperState.SyncConnected) {
                    	
                        latch.countDown();  // 连接成功
                    }
                }
            });
            latch.await();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return zk;
    }
    
    private void initCreateIndexNode(ZooKeeper zk) {
        try {
            byte[] bytes = "rabbit".getBytes();
            //zk.delete("/zk-node", -1); 
            zk.exists("/zk-node", true);
            //zk.create("/zk-node", "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
           // zk.delete("/zk-node/" + NetShareInstance.serverIp, -1); 
            //String path = zk.create("/zk-node/" + NetShareInstance.serverIp, bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            Stat stat = zk.exists(ServerDataCache.ZK_CHAT_USER_INDEX_NODE_PATH, true);    
    		if(stat == null){
    			String path1 = zk.create(ServerDataCache.ZK_CHAT_USER_INDEX_NODE_PATH, bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    			System.out.println("初始化节点:" + path1); 
    		}
            /*String path2 = zk.create(ServerDataCache.ZK_CHAT_USER_INDEX_NODE_PATH + "/2", bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            ServerDataCache.serverIdToNodePathMap.put(1+"", path1) ;  // 缓存一下
            ServerDataCache.serverIdToNodePathMap.put(2+"", path2) ;  // 缓存一下
            System.out.println("create zookeeper node path:"+path1+" data:"+data);
            System.out.println("create zookeeper node path:"+path2+" data:"+data);*/
            
        } catch (Exception e) {
        	e.printStackTrace();
        }
    }
    
    public void createUserNode( String userId, String serverId){
    	try {
    		
    		UserServerInfo userServerInfo = new UserServerInfo();
    		userServerInfo.setServerId(serverId);
    		userServerInfo.setUserId(userId);
    		String jsonData = JsonUtils.writeValueAsString(userServerInfo);
    		if(userIdToNodePathMap.get(userId) == null){
    			String nodeUserPath = zk.create(ServerDataCache.ZK_CHAT_USER_INDEX_NODE_PATH + "/" + userId
						, jsonData.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
    			logger.info("创建{}节点", nodeUserPath);
    			userIdToNodePathMap.put(userId, nodeUserPath);
    		}else{
    			Stat stat = zk.exists(userIdToNodePathMap.get(userId), true);    
        		if(stat == null){
        			try {
        				String nodeUserPath = zk.create(ServerDataCache.ZK_CHAT_USER_INDEX_NODE_PATH + "/" + userId
								, jsonData.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
						logger.info("创建{}节点", nodeUserPath);
						userIdToNodePathMap.put(userId, nodeUserPath);
						zk.setData(nodeUserPath
									, jsonData.getBytes(), -1);
					} catch (Exception e) {
						e.printStackTrace();
					}
        		}else{
        			zk.setData(ServerDataCache.ZK_CHAT_USER_INDEX_NODE_PATH + "/" + userId
    						, jsonData.getBytes(), -1);
        		}
    		}
		} catch (KeeperException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
    }
    
    public String getUserNodeInfo(String userId){
    	return userIdToServerMap.get(userId);
    }
    
}
